package top.wangjiu.reactive.rpc.server;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.internal.ObjectUtil;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import top.wangjiu.reactive.rpc.config.ServerConfiguration;
import top.wangjiu.reactive.rpc.exception.RpcException;
import top.wangjiu.reactive.rpc.handler.RequestHandler;
import top.wangjiu.reactive.rpc.model.RpcRequest;
import top.wangjiu.reactive.rpc.model.RpcResponse;
import top.wangjiu.reactive.rpc.model.RpcService;
import top.wangjiu.reactive.rpc.registration.ServerRegister;

import java.lang.reflect.Method;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;


public class ServerHander extends SimpleChannelInboundHandler<RpcRequest> {

    private static final Log log = LogFactory.getLog(ServerHander.class);

    private ServerRegister serverRegister;
    private ServerConfiguration configuration;
    private Executor executor;
    private final Set<String> processingRequest = new HashSet();
    private final Set<String> processedRequest = new HashSet();
    private final Map<String,RequestHandler> requestHandlerMap = new ConcurrentHashMap<>();

    public ServerHander(ServerConfiguration configuration){
        ObjectUtil.checkNotNull(configuration.getServerRegister(),"serverHander is null");
        this.serverRegister = configuration.getServerRegister();
        this.executor = configuration.getRequestExecutor();
        this.configuration = configuration;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("channelActive:"+ctx.channel().remoteAddress());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("channelInactive:"+ctx.channel().remoteAddress());
        requestHandlerMap.values().forEach(requestHandler -> {
            requestHandler.onDisconnect();
        });
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if(processingRequest.isEmpty()){
            ctx.channel().close().sync();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {

        if(request.isCancel() && !processingRequest.contains(request.getRequestId())){
            //消费端要取消后续处理，而服务端已经不再处理时
            return;
        }else if(processedRequest.contains(request.getRequestId()) && !processingRequest.contains(request.getRequestId())){
            log.info("ingore completed request:"+request);
            return;
        }
        processingRequest.add(request.getRequestId());
        processedRequest.add(request.getRequestId());
        if(executor!=null){
            executor.execute(()->processRequest(ctx.channel(),request));
        }else{
            processRequest(ctx.channel(),request);
        }

    }

    private RequestHandler buildRequestHandler(Channel channel,RpcRequest request) {
        RpcService service = this.serverRegister.getService(request.getService(), request.getVersion());
        if(service == null){
            return null;
        }else {
            Method method = null;
            try {
                method = service.getType().getMethod(request.getMethod(), request.getParamTypes());
                if(method==null){
                    return null;
                }
                Class<?> returnType = method.getReturnType();
                Class<? extends RequestHandler<?>> resultTypeHandler = configuration.getResultTypeHandler(returnType);
                RequestHandler<?> requestHandler = resultTypeHandler.getConstructor().newInstance();
                Method finalMethod = method;
                requestHandler.setInvoker(r-> finalMethod.invoke(service.getInstance(), r.getParams()));
                requestHandler.setResultWriter(response -> {
                    if(response.isLast()){
                        processingRequest.remove(response.getRequestId());
                    }
                    processResopnse(channel,response);
                });
                return requestHandler;
            } catch (Exception e) {
                return null;
            }
        }
    }

    private void processRequest(Channel channel,RpcRequest msg){

        RequestHandler requestHandler = requestHandlerMap.get(msg.getRequestId());
        if(requestHandler==null){
            requestHandler = buildRequestHandler(channel,msg);
            if(requestHandler!=null){
                requestHandlerMap.put(msg.getRequestId(), requestHandler);
            }
        }

        if(requestHandler == null){
            RpcResponse response = new RpcResponse(msg.getRequestId());
            response.setLast(true);
            response.setSuccess(false);
            response.setCause(new RpcException("not found service"));
            processResopnse(channel,response);
        }else{
            requestHandler.onRequest(msg);
        }

    }

    private synchronized void processResopnse(Channel channel,RpcResponse response){
        if(channel.isActive()){
            synchronized (channel){
                channel.writeAndFlush(response);
            }
        }
    }

}
